package com.hezhiqin.kafkademo;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.IntegerDeserializer;
import org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;


/**
 * @program: homeWork
 * @description:
 * @author: hezhiqin
 * @create: 2020-02-26 13:33
 */
public class HZQKafkaConsumer extends Thread{

    KafkaConsumer<Integer,String> consumer ;
    String topic;

    public HZQKafkaConsumer(String topic) {
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.223.128:9092,192.168.223.129:9092,192.168.223.130:9092");
        properties.put(ConsumerConfig.CLIENT_ID_CONFIG, "hezhiqin-consumer");
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "hezhiqin-id");
        properties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");
        properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        consumer = new KafkaConsumer<Integer, String>(properties);
        this.topic = topic;
    }

    @Override
    public void run() {

        while(true){
            consumer.subscribe(Collections.singleton(topic));
            ConsumerRecords<Integer, String> consumerRecords = consumer.poll(Duration.ofSeconds(1));
            consumerRecords.forEach(record->{
                System.out.println(record.key()+"->"+record.value()+"->"+record.offset());
            });

        }
    }


    public static void main(String[] args) {
        new HZQKafkaConsumer("kafkaTest").start();
    }


}
